package com.lee.message.core.aspectj;

import com.lee.message.core.annotation.Idempotent;
import com.lee.message.core.config.MqIdempotentProperties;
import com.lee.message.core.config.consts.Consts;
import com.lee.message.core.converter.MessageConverter;
import com.lee.message.core.config.cache.Cache;
import com.lee.message.core.config.lock.Lock;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.Order;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 消息处理抽象类，order值越小，优先级越高
 */
@Slf4j
@Order(Ordered.LOWEST_PRECEDENCE)
public abstract class AbstractMqIdempotentAspect {

    @Autowired
    protected Cache cache;

    @Autowired
    protected Lock lock;

    @Autowired
    protected MqIdempotentProperties properties;

    @Autowired
    protected MessageConverter converter;

    /**
     * 切入点配置方法
     */
    public abstract void idempotentPointCut();

    /**
     * 是否需要幂等性校验
     */
    public abstract boolean isIdempotentCheck(ProceedingJoinPoint point);

    /**
     * 获取消费组（不同组可以消费同一个消息）
     * @param point
     * @return
     */
    public abstract String consumerGroup(ProceedingJoinPoint point);

    /**
     * 获取tpoic
     * @param point
     * @return
     */
    public abstract String topic(ProceedingJoinPoint point);

    @Around("idempotentPointCut()")
    public Object around(ProceedingJoinPoint point) {
        StringBuilder storeKeyBuffer = new StringBuilder();
        StringBuilder lockKeyBuffer = new StringBuilder();

        try {
            // 是否执行幂等性检测
            boolean isIdempotentCheck = isIdempotentCheck(point);
            if(!isIdempotentCheck){
                return point.proceed();
            }

            // 消费组,topic
            String consumerGroup = consumerGroup(point);
            String topic = topic(point);

            // 获取方法,参数和返回类型
            Object[] args = point.getArgs();
            MethodSignature signature = (MethodSignature) point.getSignature();
            Method method = signature.getMethod();
            String returnTypeName = method.getReturnType().getName();

            // 获取配置注解存储时间
            Idempotent idempotent = getAnnotation(point, Idempotent.class);
            String type = properties.getType();
            Integer storeTime = properties.getStoreTime();
            TimeUnit storeTimeUnit = properties.getStoreTimeUnit();
            Integer lockTime = properties.getLockTime();
            TimeUnit lockTimeUnit = properties.getLockTimeUnit();

            if(idempotent != null){
                if(idempotent.storeTime() > 0){
                    storeTime = idempotent.storeTime();
                    storeTimeUnit = idempotent.storeTimeUnit();
                }

                if(idempotent.lockTime() > 0){
                    lockTime = idempotent.lockTime();
                    lockTimeUnit = idempotent.lockTimeUnit();
                }

                if(StringUtils.hasText(idempotent.type())){
                    type = idempotent.type();
                }
            }

            // 方法返回值仅支持 boolean void 返回值
            boolean isVoid = returnTypeName.equalsIgnoreCase("void");
            if (!isVoid && !returnTypeName.equalsIgnoreCase("boolean")) {
                throw new RuntimeException("方法返回类型必须void或boolean");
            }

            // 需要从参数对象中获取key
            if(args == null || args[0] == null){
                throw new RuntimeException("方法必须有参数");
            }

            String msgId = converter.getUniqueKey(args[0]);
            if(!StringUtils.hasText(msgId)){
                throw new RuntimeException("msgId不能为空");
            }

            storeKeyBuffer.append(properties.getStoreKey());
            append(storeKeyBuffer, consumerGroup);
            append(storeKeyBuffer, topic);
            append(storeKeyBuffer, type);
            storeKeyBuffer.append(msgId);
            String storeKey = storeKeyBuffer.toString();

            lockKeyBuffer.append(properties.getLockKey());
            append(lockKeyBuffer, consumerGroup);
            append(lockKeyBuffer, topic);
            append(lockKeyBuffer, type);
            lockKeyBuffer.append(msgId);

            log.info("消费消息" + storeKey);

            // 已经消费则正常结束
            if (exitKey(storeKey)) {
                log.warn("消息" + storeKey + "已经消费");
                return isVoid ? null : true;
            }

            // 加锁报错则重新消费
            boolean isLock = lock.tryLock(lockKeyBuffer.toString(), lockTime, lockTimeUnit);
            if (!isLock) {
                log.warn("消息" + storeKey + "正在被消费");
                throw new RuntimeException("消息" + storeKey + "正在被消费");
            }

            Object proceed = point.proceed();

            // 添加记录到redis或DB
            // 方式1：使用redis存储消费信息；如果有事务，会存在缓存添加成功而数据消费失败极端情况；没有事务，会存在数据消费成功，缓存添加失败情况（有风险）
            // 方式2：使用db存储消费信息，幂等数据和业务数据一起入库，同时成功或者失败
            cache.setCacheObject(storeKey, Consts.DEFAULT_VAUE, storeTime, storeTimeUnit);
            return proceed;
        } catch (RuntimeException e) {
            log.error("幂等处理失败", e);
            throw e;
        } catch (Throwable throwable) {
            log.error("幂等处理失败", throwable);
            throw new RuntimeException(throwable);
        } finally {
            // 释放锁
            unlock(lockKeyBuffer.toString());
        }
    }

    /**
     * 添加数据，带分隔符
     * @param stringBuilder
     * @param append
     */
    protected void append(StringBuilder stringBuilder,String append){
        if(StringUtils.hasText(append)){
            stringBuilder.append(append);
            stringBuilder.append(":");
        }
    }

    /**
     * 先获取方法注解，方法没有则获取类
     * @param point
     * @param clazz
     * @param <T>
     * @return
     */
    protected  <T> T getAnnotation(ProceedingJoinPoint point,Class clazz) {
        MethodSignature signature = (MethodSignature) point.getSignature();
        T t = (T) AnnotationUtils.findAnnotation(signature.getMethod(), clazz);
        if (Objects.nonNull(t)) {
            return t;
        }

        return (T) AnnotationUtils.findAnnotation(signature.getDeclaringType(), clazz);
    }

    /**
     * 释放锁
     * @param lockKey
     */
    protected void unlock(String lockKey) {
        if(StringUtils.hasText(lockKey)){
            lock.unlock(lockKey);
        }
    }

    /**
     * 判断是否存在锁
     * @param key
     * @return
     */
    protected boolean exitKey(String key) {
        // 判断是否存在
        checkKey(key);
        return lock.exists(key);
    }

    /**
     * 检查key
     * @param key
     */
    protected void checkKey(String key){
        if(!StringUtils.hasText(key)){
            throw new RuntimeException("key不能为空");
        }
    }

}
